package com.li.flink.datastream

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object StreamingWordCount {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //连接socket
    val text = env.socketTextStream("bigdata04", 9001)

    //处理数据
    import org.apache.flink.api.scala._
    val wordCount = text.flatMap(_.split(" ")) //将每一行数据根据空格切分单词
      .map((_, 1)) // 转为tuple2形式(单词,1)
      .keyBy(tup => tup._1) //根据tuple2第一列分组
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) //每隔两秒执行计算一次接收的数据
      .sum(1) //sum or reduce 聚合

    wordCount.print().setParallelism(1) //使用一个线程执行打印操作

    //执行
    env.execute("StreamingWordCount")
  }
}
